package com.atguigu.bigdata.gmall.realtime.app.dwd;/**
 * @Classname Dwd_03_DwdCourseEval
 * @Description TODO
 * @Date 2022/10/25 14:33
 * @Created by 86152
 */


import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author: Lime
 * @description 课程评价明细表
 * @date: 2022/10/25 14:33
 */
public class Dwd_03_DwdCourseEval extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_03_DwdCourseEval().init(
                3003,
                2,
                "Dwd_03_DwdCourseEval"
        );
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //sql装
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));

        //1.读取ods_db11
        readOdsDb(tEnv,"Dwd_03_DwdCourseEval");
//        tEnv.executeSql("select * from ods_db").print();




        readBaseDicCourse(tEnv);
        //测试
//        tEnv.executeSql("select * from course_info").print();


        Table courseReview = tEnv.sqlQuery("select\n" +
                " data['id'] id,\n" +
                " data['user_id'] user_id,\n" +
                " cast(data['course_id'] as bigint) course_id,\n" +
                " data['review_stars'] review_stars,\n" +
                " data['create_time'] create_time,\n" +
                " ts,\n" +
                " pt\n" +
                "from ods_db\n" +
                "where `database`='gmall2022' and `table`='review_info' and `type`='insert'");
        tEnv.createTemporaryView("courseReview",courseReview);


//        tEnv.executeSql("select * from courseReview").print();


        //4.两张表的join
        Table result1 = tEnv.sqlQuery("select\n" +
                " cR.id,\n" +
                " cR.course_id,\n" +
                " cI.course_name,\n" +
                " cR.user_id,\n" +
                " cR.review_stars,\n" +
                " cR.ts\n" +
                "from courseReview cR join course_info for system_time as of cR.pt as cI on cR.course_id = cI.id");
//        tEnv.createTemporaryView("result1",result1);
//        tEnv.executeSql("select * from result1").print();



        // 7. 写出到 kafka 中
        tEnv.executeSql("create table dwd_Eval_Course(\n" +
                " id string,\n" +
                " course_id bigint,\n" +
                " course_name string,\n" +
                " user_id string,\n" +
                " review_stars string,\n" +
                " ts bigint,\n" +
                " primary key(id) not enforced\n" +
                ")" + SQLUtil.getUpsertKafkaSink(Constant.TOPIC_DWD_EVAL_COURSE));

        result1.executeInsert("dwd_Eval_Course");



    }
}
